/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */

package homomorphism.NewAPI;


import java.io.IOException;
import java.lang.reflect.*;
import java.util.ArrayList;
import java.util.logging.Level;
import java.util.logging.Logger;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.mapreduce.Reducer;

import utility.LongPair;
import homomorphism.HomomorphismBase;
import java.util.Iterator;

/**
 *
 * @author takeshi
 */
public class LHCominber<InVal, OtVal> extends Reducer<LongPair, InVal, LongPair, InVal> {

    private static final Log LOG = LogFactory.getLog(LHCominber.class);
    private static final String userAppProperty = HomomorphismBase.UserClassNameProperty;
    private Class FoldingType;
    private Class UnFoldingType;
    private String UserClassName;

    @Override
    public void setup(Context context) {
        UserClassName = context.getConfiguration().get(userAppProperty);
        FoldingType = context.getMapOutputValueClass();
        UnFoldingType = context.getOutputValueClass();
   
    }

    //to be delete
      @Deprecated
    public OtVal fold(ArrayList<InVal> values) throws Exception {
        if (UserClassName == null || UserClassName.isEmpty()) {
            throw new IOException("1null  fold:UserClassName\n");
        }
        try {
            Class UserApp = Class.forName(UserClassName);
            Class partypes[] = new Class[1];
            partypes[0] = values.getClass();
            Method _fold = UserApp.getMethod("fold", partypes);

            Object arglist[] = new Object[1];
            arglist[0] = values;
            Constructor ct = UserApp.getConstructor(); //partypes
            Object methobj = ct.newInstance();
            OtVal rst = (OtVal) _fold.invoke(methobj, arglist);
            return rst;

        } catch (ClassNotFoundException ee) {
            LOG.error(ee.getMessage());
        }
        return null;
    }

        public OtVal fold(Iterator<InVal> values) throws Exception {

        if (UserClassName == null || UserClassName.isEmpty()) {
            throw new IOException("1null  fold:UserClassName\n");
        }
        try {
            Class UserApp = Class.forName(UserClassName);
            Class partypes[] = new Class[1];
            partypes[0] = Iterator.class;
            Method _fold = UserApp.getMethod("fold", partypes);

            // Object arglist[] = new Object[1];
            Iterator arglist[] = new Iterator[1];
            arglist[0] = values;
            Constructor ct = UserApp.getConstructor(); //partypes
            Object methobj = ct.newInstance();
            OtVal rst = (OtVal) _fold.invoke(methobj, arglist);
            return rst;
        } catch (ClassNotFoundException ee) {
            // Logger.getLogger(ReducerFor3rdLstHomoThm1stPh.class.getName()).log(Level.SEVERE, null, ee);
            LOG.error(ee.getMessage() + values.getClass().getName());
        } catch (IllegalArgumentException ex) {
            Logger.getLogger(ReducerFor3rdLstHomoThm1stPh.class.getName()).log(Level.SEVERE, null, ex);
            LOG.error(ex.getMessage() + values.getClass().getName() + " Iterator<InVal> is " + values.getClass().toString());
        }
        return null;

    }
        
    public ArrayList<InVal> unfold(OtVal value) throws Exception {
        Class UserApp = Class.forName(UserClassName);
        Class partypes[] = new Class[1];
        partypes[0] = value.getClass();//UnFoldingType;
        Method _unfold = UserApp.getMethod("unfold", partypes);

        Object arglist[] = new Object[1];
        arglist[0] = value;
        Constructor ct = UserApp.getConstructor(); //partypes
        Object methobj = ct.newInstance();
        ArrayList<InVal> rst = (ArrayList<InVal>) _unfold.invoke(methobj, arglist);
        return rst;
    }

    private OtVal LeftWards(Iterable<InVal> values) throws Exception {

        if (values.iterator() != null) {
            return fold(values.iterator());
        }
        
        OtVal outVal = null;
        ArrayList<InVal> tmpList = new ArrayList<InVal>();
        for (InVal e : values) {
            if (outVal != null) {
                for (InVal ee : unfold(outVal)) {
                    tmpList.add(ee);
                }
            }
            tmpList.add(e);
            outVal = fold(tmpList);
            tmpList.clear();
        }
        return outVal;
    }

    @Override
    public void reduce(LongPair key, Iterable<InVal> values, Context context) throws IOException, InterruptedException {
        try {
            OtVal aggreted = LeftWards(values);
            int count =0;
            for (InVal e : unfold(aggreted)) {
                LongPair newkey = new LongPair( key.getFirst(), key.getSecond()+count );
                 context.write(newkey, e);
            }
        } catch (Exception ex) {
            Logger.getLogger(LHCominber.class.getName()).log(Level.SEVERE, null, ex);
        }

    }
}
